package org.zjt.mr.weather;


import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;

public class RunJob {

    public static void main1() {
        Configuration config = new Configuration();
        config.set("fs.defaultFS", "hdfs://master:9000");
        config.set("yarn.resourcemanager.hostname", "master");
        System.setProperty("HADOOP_USER_NAME", "root");    //设置用户的角色
//		config.set("mapred.jar", "C:\\Users\\Administrator\\Desktop\\wc.jar");
//		config.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");
        try {
            FileSystem fs = FileSystem.get(config);

            Job job = Job.getInstance(config);
            job.setJarByClass(RunJob.class);

            job.setJobName("weather");

            job.setMapperClass(WeatherMapper.class);
            job.setReducerClass(WeatherReducer.class);
            job.setMapOutputKeyClass(MyKey.class);
            job.setMapOutputValueClass(DoubleWritable.class);

            job.setPartitionerClass(MyPartitioner.class);
            job.setSortComparatorClass(MySort.class);
            job.setGroupingComparatorClass(MyGroup.class);   //将相同的mapreduce中数据分组聚合，同样属性的数据到相同的reduce中。

            job.setNumReduceTasks(3);

            job.setInputFormatClass(KeyValueTextInputFormat.class);

            FileInputFormat.addInputPath(job, new Path("/usr/input/weather"));

            Path outpath = new Path("/usr/output/weather");
            if (fs.exists(outpath)) {
                fs.delete(outpath, true);
            }
            FileOutputFormat.setOutputPath(job, outpath);

            boolean f = job.waitForCompletion(true);
            if (f) {
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //key：每行第一个隔开符左边为key，右边为value( hadoop的默认分隔符为制表符 /t )
    static class WeatherMapper extends Mapper<Text, Text, MyKey, DoubleWritable> {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        NullWritable v = NullWritable.get();

        protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
            try {
                Date date = sdf.parse(key.toString());
                Calendar c = Calendar.getInstance();
                c.setTime(date);
                int year = c.get(Calendar.YEAR);
                int month = c.get(Calendar.MONTH);

                double hot = Double.parseDouble(value.toString().substring(0, value.toString().lastIndexOf("c")));
                MyKey k = new MyKey();
                k.setYear(year);
                k.setMonth(month);
                k.setHot(hot);
                context.write(k, new DoubleWritable(hot));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    static class WeatherReducer extends Reducer<MyKey, DoubleWritable, Text, NullWritable> {
        protected void reduce(MyKey arg0, Iterable<DoubleWritable> arg1,
                              Context arg2)
                throws IOException, InterruptedException {
            int i = 0;
            for (DoubleWritable v : arg1) {
                i++;
                String msg = arg0.getYear() + "\t" + arg0.getMonth() + "\t" + v.get();
                arg2.write(new Text(msg), NullWritable.get());
                if (i == 3) {
                    break;
                }
            }
        }
    }


    public static void updateDir() throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://master:9000/"), config, "root");
        Path file = new Path("/usr/input/weather/weather.txt");
        FSDataOutputStream out = fs.create(file);
//		IOUtils.copy(input, output)
        org.apache.hadoop.io.IOUtils.copyBytes(new FileInputStream("/Users/zhangjuntao/IdeaProjects/myproject/hadoop-demo/hdfs-demo/src/main/resources/data/weather"), out, config);
        fs.close();
    }


    public static void printResult() throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://master:9000/"), config, "root");
        Path file = new Path("/usr/output/weather/");
        FileStatus[] fss = fs.listStatus(file);
        for (FileStatus fileName : fss) {
            System.out.println(fileName.getPath().getName() + ":");
            BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(fileName.getPath())));
            String line;
            line = br.readLine();
            while (line != null) {
                System.out.println("\t" + line);
                line = br.readLine();
            }
        }
        fs.close();
    }

    public static void main(String[] args) throws Exception {

        // TODO: 2017/5/23 将数组转化为字符串+分隔符
        String[] array = {"asd", "wqwqw", "zhang", "wang"};
        System.out.println(StringUtils.join(array, ","));
        System.out.println(StringUtils.join(Arrays.asList(array), ","));
        printResult();
    }
}
